/**
 * 抖音关注：程序员三丙
 * 知识星球：https://t.zsxq.com/j9b21
 */
package sanbing.example.dylike.infrastructure.queue.provider;

import jakarta.annotation.PreDestroy;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
import sanbing.example.dylike.infrastructure.queue.QueueAdmin;
import sanbing.example.dylike.infrastructure.queue.QueueConsumer;
import sanbing.example.dylike.infrastructure.queue.QueueMsg;
import sanbing.example.dylike.infrastructure.queue.QueueProducer;
import sanbing.example.dylike.infrastructure.queue.discovery.ServiceInfoProvider;
import sanbing.example.dylike.infrastructure.queue.kafka.*;
import sanbing.example.dylike.infrastructure.queue.rpc.DefaultQueueRequestTemplate;
import sanbing.example.dylike.infrastructure.queue.rpc.QueueRequestTemplate;
import sanbing.example.dylike.infrastructure.queue.settings.QueueCoreSettings;
import sanbing.example.dylike.infrastructure.queue.settings.QueueRpcSettings;

@Component
@ConditionalOnExpression("'${queue.type:null}'=='kafka'")
public class KafkaQueueFactory implements QueueFactory {

    private final KafkaSettings kafkaSettings;
    private final QueueCoreSettings coreSettings;
    private final QueueRpcSettings rpcSettings;
    private final KafkaConsumerStatsService consumerStatsService;
    private final ServiceInfoProvider serviceInfoProvider;

    private final QueueAdmin coreAdmin;
    private final QueueAdmin rpcRequestAdmin;
    private final QueueAdmin rpcResponseAdmin;

    public KafkaQueueFactory(KafkaSettings kafkaSettings,
                             ServiceInfoProvider serviceInfoProvider,
                             QueueCoreSettings coreSettings,
                             QueueRpcSettings rpcSettings,
                             KafkaConsumerStatsService consumerStatsService,
                             KafkaTopicConfigs kafkaTopicConfigs) {
        this.kafkaSettings = kafkaSettings;
        this.serviceInfoProvider = serviceInfoProvider;
        this.coreSettings = coreSettings;
        this.rpcSettings = rpcSettings;
        this.consumerStatsService = consumerStatsService;

        this.coreAdmin = new KafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs());
        this.rpcRequestAdmin = new KafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRpcRequestConfigs());
        this.rpcResponseAdmin = new KafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRpcResponseConfigs());
    }


    @Override
    public QueueProducer createToClickHouseMsgProducer() {
        KafkaProducerTemplate.KafkaProducerTemplateBuilder requestBuilder = KafkaProducerTemplate.builder();
        requestBuilder.settings(kafkaSettings);
        requestBuilder.clientId("like-clickhouse-producer-" + serviceInfoProvider.getServiceId());
        requestBuilder.topic(coreSettings.getClickhouseTopic());
        requestBuilder.admin(coreAdmin);
        return requestBuilder.build();
    }

    @Override
    public QueueRequestTemplate createSeckillRpcRequestTemplate() {
        KafkaProducerTemplate.KafkaProducerTemplateBuilder requestBuilder = KafkaProducerTemplate.builder();
        requestBuilder.settings(kafkaSettings);
        requestBuilder.clientId("seckill-api-request-" + serviceInfoProvider.getServiceId());
        requestBuilder.topic(rpcSettings.getSeckillRequestsTopic());
        requestBuilder.admin(rpcRequestAdmin);

        KafkaConsumerTemplate.KafkaConsumerTemplateBuilder responseBuilder = KafkaConsumerTemplate.builder();
        responseBuilder.settings(kafkaSettings);
        responseBuilder.topic(rpcSettings.getSeckillResponsesTopic() + "." + serviceInfoProvider.getServiceId());
        responseBuilder.clientId("seckill-api-response-" + serviceInfoProvider.getServiceId());
        responseBuilder.groupId("rpc-api-seckill-" + serviceInfoProvider.getServiceId());
        responseBuilder.decoder(msg -> new QueueMsg(msg.getKey(), msg.getData(), msg.getHeaders()));
        responseBuilder.admin(rpcResponseAdmin);
        responseBuilder.statsService(consumerStatsService);

        DefaultQueueRequestTemplate.DefaultQueueRequestTemplateBuilder templateBuilder = DefaultQueueRequestTemplate.builder();
        templateBuilder.queueAdmin(rpcResponseAdmin);
        templateBuilder.requestTemplate(requestBuilder.build());
        templateBuilder.responseTemplate(responseBuilder.build());
        templateBuilder.maxPendingRequests(rpcSettings.getMaxPendingRequests());
        templateBuilder.maxRequestTimeout(rpcSettings.getMaxRequestsTimeout());
        templateBuilder.pollInterval(rpcSettings.getResponsePollInterval());
        return templateBuilder.build();
    }



    @Override
    public QueueConsumer createSeckillRpcRequestConsumer() {
        KafkaConsumerTemplate.KafkaConsumerTemplateBuilder consumerBuilder = KafkaConsumerTemplate.builder();
        consumerBuilder.settings(kafkaSettings);
        consumerBuilder.topic(rpcSettings.getSeckillRequestsTopic());
        consumerBuilder.clientId("rpc-seckill-request-consumer-" + serviceInfoProvider.getServiceId());
        consumerBuilder.groupId("rpc-seckill-request-consumer");
        consumerBuilder.decoder(msg -> new QueueMsg(msg.getKey(), msg.getData(), msg.getHeaders()));
        consumerBuilder.admin(rpcRequestAdmin);
        consumerBuilder.statsService(consumerStatsService);
        return consumerBuilder.build();
    }

    @Override
    public QueueProducer createSeckillResponseProducer() {
        KafkaProducerTemplate.KafkaProducerTemplateBuilder requestBuilder = KafkaProducerTemplate.builder();
        requestBuilder.settings(kafkaSettings);
        requestBuilder.clientId("rpc-producer-" + serviceInfoProvider.getServiceId());
        requestBuilder.topic(rpcSettings.getSeckillResponsesTopic());
        requestBuilder.admin(rpcResponseAdmin);
        return requestBuilder.build();
    }


    @PreDestroy
    private void destroy() {
        if (coreAdmin != null) {
            coreAdmin.destroy();
        }
        if (rpcRequestAdmin != null) {
            rpcRequestAdmin.destroy();
        }
        if (rpcResponseAdmin != null) {
            rpcResponseAdmin.destroy();
        }
    }
}
